﻿using System;
using System.Threading.Tasks;
using Confluent.Kafka;
using DelayedQueue.Models;

namespace DelayedQueue.Kafka
{
    /// <summary>
    /// Kafka生产者，负责发送延时任务消息
    /// </summary>
    public class KafkaProducer : IDisposable
    {
        private readonly IProducer<string, string> _producer;
        private readonly string _topic;
        private bool _isDisposed;

        public KafkaProducer(string bootstrapServers, string topic)
        {
            if (string.IsNullOrEmpty(bootstrapServers))
                throw new ArgumentException("Kafka服务器地址不能为空", nameof(bootstrapServers));

            if (string.IsNullOrEmpty(topic))
                throw new ArgumentException("主题名称不能为空", nameof(topic));

            _topic = topic;

            var config = new ProducerConfig
            {
                BootstrapServers = bootstrapServers,
                Acks = Acks.All,
                //RetryCount = 3
            };

            _producer = new ProducerBuilder<string, string>(config).Build();
        }

        /// <summary>
        /// 发送延时任务到Kafka
        /// </summary>
        public async Task<DeliveryResult<string, string>> SendDelayedTaskAsync(DelayedTask task)
        {
            if (task == null)
                throw new ArgumentNullException(nameof(task));

            var message = new Message<string, string>
            {
                Key = task.TaskId,
                Value = task.ToJson()
            };

            try
            {
                return await _producer.ProduceAsync(_topic, message);
            }
            catch (ProduceException<string, string> ex)
            {
                Console.WriteLine($"发送消息失败: {ex.Error.Reason}");
                throw;
            }
        }

        public void Dispose()
        {
            Dispose(true);
            GC.SuppressFinalize(this);
        }

        protected virtual void Dispose(bool disposing)
        {
            if (_isDisposed) return;

            if (disposing)
            {
                // 确保所有消息都已发送
                _producer?.Flush(TimeSpan.FromSeconds(10));
                _producer?.Dispose();
            }

            _isDisposed = true;
        }

        ~KafkaProducer()
        {
            Dispose(false);
        }
    }
}
